package com.shujia.card

import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.util.{CarUtil, SparkStreamTool}
import org.apache.spark.streaming.{Duration, Durations}
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

object RealTimeRoadAvgSpeed extends SparkStreamTool {
  override def run(args: Array[String]): Unit = {

    /**
      *
      * 1.3 实时统计道路拥堵情况-  实时计算每个道路平均速度，计算最近5分钟每个一分钟计算一次
      */

    //读取数据
    val carsDS: DStream[CarUtil.Car] = CarUtil.loadKafkaCar(ssc, "RealTimeRoadAvgSpeed")


    val kvDS: DStream[(Long, (Double, Int))] = carsDS.map(car => {
      val road: Long = car.road_id

      val speed: Double = car.speed

      (road, (speed, 1))
    })

    val reduce = (x: (Double, Int), y: (Double, Int)) => {
      val sumSpeed: Double = x._1 + y._1
      val sumNum: Int = x._2 + y._2
      (sumSpeed, sumNum)
    }


    //统计每一个道路最新5分钟总的车速和总的车流量
    val speedAndNum: DStream[(Long, (Double, Int))] = kvDS.reduceByKeyAndWindow(
      reduce,
      Durations.minutes(5),
      Durations.minutes(1)
    )


    //计算平均车速
    val avgSpeedDS: DStream[(Long, Double)] = speedAndNum.map {
      case (road: Long, (sumSpeed: Double, num: Int)) =>
        (road, sumSpeed / num)
    }

    //将数据保存到redis中

    avgSpeedDS.foreachRDD(rdd => {
      rdd.foreachPartition(iter => {

        //奖励redis链接
        val jedis = new Jedis("master", 6379)

        val date = new Date
        val format = new SimpleDateFormat("yyyyMMddHHmmss")
        val time: String = format.format(date)


        iter.foreach {
          case (road: Long, avgSpeed: Double) =>

            val key: String = "RealTimeRoadAvgSpeed:" + road + ":" + time

            jedis.set(key, avgSpeed.toString)

            //不需要保留太久的历史数据，可以使用redis的 TTL 自动删除数据
            jedis.expire(key, 36000)
        }

        jedis.close()
      })
    })

  }
}
